/*-------------------------------------------------------------------------
 *
 * walreceiverfuncs.c
 *
 * 此文件包含启动进程与 walreceiver 进程通信所使用的函数。
 * 实现 walreceiver 本身的函数在 walreceiver.c 中。
 *
 * Portions Copyright (c) 2010-2022, PostgreSQL Global Development Group
 *
 * IDENTIFICATION
 *	  src/backend/replication/walreceiverfuncs.c
 *
 *-------------------------------------------------------------------------
 */
#include "postgres.h"

#include <sys/stat.h>
#include <sys/time.h>
#include <time.h>
#include <unistd.h>
#include <signal.h>

#include "access/xlog_internal.h"
#include "access/xlogrecovery.h"
#include "pgstat.h"
#include "postmaster/startup.h"
#include "replication/walreceiver.h"
#include "storage/pmsignal.h"
#include "storage/shmem.h"
#include "utils/timestamp.h"

WalRcvData *WalRcv = NULL;

/*
 * 等待 walreceiver 启动的时间，在请求 postmaster 启动它后，以秒为单位。
 */
#define WALRCV_STARTUP_TIMEOUT 10

/* 报告 WalRcvShmemInit 所需的共享内存空间 */
Size WalRcvShmemSize(void)
{
	Size		fc_size = 0;

	fc_size = add_size(fc_size, sizeof(WalRcvData));

	return fc_size;
}

/* 分配并初始化与 walreceiver 相关的共享内存 */
void WalRcvShmemInit(void)
{
	bool		fc_found;

	WalRcv = (WalRcvData *)
		ShmemInitStruct("Wal Receiver Ctl", WalRcvShmemSize(), &fc_found);

	if (!fc_found)
	{
		/* 第一次通过，所以初始化 */
		MemSet(WalRcv, 0, WalRcvShmemSize());
		WalRcv->walRcvState = WALRCV_STOPPED;
		ConditionVariableInit(&WalRcv->walRcvStoppedCV);
		SpinLockInit(&WalRcv->mutex);
		pg_atomic_init_u64(&WalRcv->writtenUpto, 0);
		WalRcv->latch = NULL;
	}
}

/* walreceiver 是否正在运行（或启动中）？ */
bool WalRcvRunning(void)
{
	WalRcvData *fc_walrcv = WalRcv;
	WalRcvState fc_state;
	pg_time_t	fc_startTime;

	SpinLockAcquire(&fc_walrcv->mutex);

	fc_state = fc_walrcv->walRcvState;
	fc_startTime = fc_walrcv->startTime;

	SpinLockRelease(&fc_walrcv->mutex);

	/*
	 * 如果 walreceiver 启动花费的时间过长，则放弃。将状态设置为 STOPPED 确保
	 * 如果 walreceiver 最终确实启动，它会看到不应该运行并在不执行任何操作的情况下终止。
	 */
	if (fc_state == WALRCV_STARTING)
	{
		pg_time_t	fc_now = (pg_time_t) time(NULL);

		if ((fc_now - fc_startTime) > WALRCV_STARTUP_TIMEOUT)
		{
			bool		fc_stopped = false;

			SpinLockAcquire(&fc_walrcv->mutex);
			if (fc_walrcv->walRcvState == WALRCV_STARTING)
			{
				fc_state = fc_walrcv->walRcvState = WALRCV_STOPPED;
				fc_stopped = true;
			}
			SpinLockRelease(&fc_walrcv->mutex);

			if (fc_stopped)
				ConditionVariableBroadcast(&fc_walrcv->walRcvStoppedCV);
		}
	}

	if (fc_state != WALRCV_STOPPED)
		return true;
	else
		return false;
}

/*
 * walreceiver 是否正在运行和流式传输（或至少尝试连接，或正在启动）？
 */
bool WalRcvStreaming(void)
{
	WalRcvData *fc_walrcv = WalRcv;
	WalRcvState fc_state;
	pg_time_t	fc_startTime;

	SpinLockAcquire(&fc_walrcv->mutex);

	fc_state = fc_walrcv->walRcvState;
	fc_startTime = fc_walrcv->startTime;

	SpinLockRelease(&fc_walrcv->mutex);

	/*
	 * 如果 walreceiver 启动花费的时间过长，则放弃。将状态设置为 STOPPED 确保
	 * 如果 walreceiver 最终确实启动，它会看到不应该运行并在不执行任何操作的情况下终止。
	 */
	if (fc_state == WALRCV_STARTING)
	{
		pg_time_t	fc_now = (pg_time_t) time(NULL);

		if ((fc_now - fc_startTime) > WALRCV_STARTUP_TIMEOUT)
		{
			bool		fc_stopped = false;

			SpinLockAcquire(&fc_walrcv->mutex);
			if (fc_walrcv->walRcvState == WALRCV_STARTING)
			{
				fc_state = fc_walrcv->walRcvState = WALRCV_STOPPED;
				fc_stopped = true;
			}
			SpinLockRelease(&fc_walrcv->mutex);

			if (fc_stopped)
				ConditionVariableBroadcast(&fc_walrcv->walRcvStoppedCV);
		}
	}

	if (fc_state == WALRCV_STREAMING || fc_state == WALRCV_STARTING ||
		fc_state == WALRCV_RESTARTING)
		return true;
	else
		return false;
}

/*
 * 停止 walreceiver（如果正在运行）并等待它终止。
 * 由 Startup 进程执行。
 */
void ShutdownWalRcv(void)
{
	WalRcvData *fc_walrcv = WalRcv;
	pid_t		fc_walrcvpid = 0;
	bool		fc_stopped = false;

	/*
	 * 请求 walreceiver 停止。walreceiver 完成后将切换到 WALRCV_STOPPED 
	 * 模式，并且还会请求 postmaster 不要重新启动自己。
	 */
	SpinLockAcquire(&fc_walrcv->mutex);
	switch (fc_walrcv->walRcvState)
	{
		case WALRCV_STOPPED:
			break;
		case WALRCV_STARTING:
			fc_walrcv->walRcvState = WALRCV_STOPPED;
			fc_stopped = true;
			break;

		case WALRCV_STREAMING:
		case WALRCV_WAITING:
		case WALRCV_RESTARTING:
			fc_walrcv->walRcvState = WALRCV_STOPPING;
			/* fall through */
		case WALRCV_STOPPING:
			fc_walrcvpid = fc_walrcv->pid;
			break;
	}
	SpinLockRelease(&fc_walrcv->mutex);

	/* 不必要但一致。 */
	if (fc_stopped)
		ConditionVariableBroadcast(&fc_walrcv->walRcvStoppedCV);

	/*
	 * 如果 walreceiver 仍在运行，则向其发送信号。
	 */
	if (fc_walrcvpid != 0)
		kill(fc_walrcvpid, SIGTERM);

	/*
	 * 等待 walreceiver 通过将状态设置为 WALRCV_STOPPED 来确认其终止。
	 */
	ConditionVariablePrepareToSleep(&fc_walrcv->walRcvStoppedCV);
	while (WalRcvRunning())
		ConditionVariableSleep(&fc_walrcv->walRcvStoppedCV,
							   WAIT_EVENT_WAL_RECEIVER_EXIT);
	ConditionVariableCancelSleep();
}

/*
 * 请求 postmaster 启动 walreceiver。
 *
 * "recptr" 指示流式传输应该开始的位置。"conninfo"
 * 是要使用的 libpq 连接字符串。"slotname" 是可选的，指要获取的
 * 复制槽的名称。"create_temp_slot" 指示在未给出 "slotname" 时创建
 * 临时槽。
 *
 * WAL 接收器不会直接加载用于连接到主服务器的 GUC 参数，而是依赖
 * 该例程调用者传递的值。因此，任何新参数的添加应通过此代码路径进行。
 */
void RequestXLogStreaming(TimeLineID fc_tli, XLogRecPtr fc_recptr, const char *fc_conninfo,
					 const char *fc_slotname, bool fc_create_temp_slot)
{
	WalRcvData *fc_walrcv = WalRcv;
	bool		fc_launch = false;
	pg_time_t	fc_now = (pg_time_t) time(NULL);
	Latch	   *fc_latch;

	/*
	 * 我们始终从段的开头开始。这可以防止由于 XLOG 流式传输而创建的损坏
	 * 段（即，段前半部分没有记录），如果该段被存档，可能会在之后造成麻烦。
	 */
	if (XLogSegmentOffset(fc_recptr, wal_segment_size) != 0)
		fc_recptr -= XLogSegmentOffset(fc_recptr, wal_segment_size);

	SpinLockAcquire(&fc_walrcv->mutex);

	/* 如果我们尝试重新启动它，它最好被停止 */
	Assert(fc_walrcv->walRcvState == WALRCV_STOPPED ||
		   fc_walrcv->walRcvState == WALRCV_WAITING);

	if (fc_conninfo != NULL)
		strlcpy((char *) fc_walrcv->conninfo, fc_conninfo, MAXCONNINFO);
	else
		fc_walrcv->conninfo[0] = '\0';

	/*
	 * 如果存在，则使用配置的复制槽，并忽略 create_temp_slot 的值，因为
	 * 槽名称应为持久性。否则，使用 create_temp_slot 来确定此 WAL
	 * 接收器是否应自己创建临时槽并使用它，或者不使用。
	 */
	if (fc_slotname != NULL && fc_slotname[0] != '\0')
	{
		strlcpy((char *) fc_walrcv->slotname, fc_slotname, NAMEDATALEN);
		fc_walrcv->is_temp_slot = false;
	}
	else
	{
		fc_walrcv->slotname[0] = '\0';
		fc_walrcv->is_temp_slot = fc_create_temp_slot;
	}

	if (fc_walrcv->walRcvState == WALRCV_STOPPED)
	{
		fc_launch = true;
		fc_walrcv->walRcvState = WALRCV_STARTING;
	}
	else
		fc_walrcv->walRcvState = WALRCV_RESTARTING;
	fc_walrcv->startTime = fc_now;

	/*
	 * 如果这是 walreceiver 的首次启动（在此时间线上），
	 * 则将 flushedUpto 和 latestChunkStart 初始化为起始点。
	 */
	if (fc_walrcv->receiveStart == 0 || fc_walrcv->receivedTLI != fc_tli)
	{
		fc_walrcv->flushedUpto = fc_recptr;
		fc_walrcv->receivedTLI = fc_tli;
		fc_walrcv->latestChunkStart = fc_recptr;
	}
	fc_walrcv->receiveStart = fc_recptr;
	fc_walrcv->receiveStartTLI = fc_tli;

	fc_latch = fc_walrcv->latch;

	SpinLockRelease(&fc_walrcv->mutex);

	if (fc_launch)
		SendPostmasterSignal(PMSIGNAL_START_WALRECEIVER);
	else if (fc_latch)
		SetLatch(fc_latch);
}

/*
 * 返回walreceiver已刷新后的最后+1字节位置。
 *
 * 可选地，返回之前的chunk开始位置，即在最近一次walreceiver刷新周期中
 * 写入的第一个字节。对该值不感兴趣的调用者可以为latestChunkStart传递NULL。
 * receiveTLI同样适用。
 */
XLogRecPtr GetWalRcvFlushRecPtr(XLogRecPtr *fc_latestChunkStart, TimeLineID *fc_receiveTLI)
{
	WalRcvData *fc_walrcv = WalRcv;
	XLogRecPtr	fc_recptr;

	SpinLockAcquire(&fc_walrcv->mutex);
	fc_recptr = fc_walrcv->flushedUpto;
	if (fc_latestChunkStart)
		*fc_latestChunkStart = fc_walrcv->latestChunkStart;
	if (fc_receiveTLI)
		*fc_receiveTLI = fc_walrcv->receivedTLI;
	SpinLockRelease(&fc_walrcv->mutex);

	return fc_recptr;
}

/*
 * 返回walreceiver已写入的最后+1字节位置。
 * 这在不加锁的情况下返回最近写入的值。
 */
XLogRecPtr GetWalRcvWriteRecPtr(void)
{
	WalRcvData *fc_walrcv = WalRcv;

	return pg_atomic_read_u64(&fc_walrcv->writtenUpto);
}

/*
 * 返回复制应用延迟，单位为毫秒，如果应用延迟信息不可用，则返回-1。
 */
int GetReplicationApplyDelay(void)
{
	WalRcvData *fc_walrcv = WalRcv;
	XLogRecPtr	fc_receivePtr;
	XLogRecPtr	fc_replayPtr;
	TimestampTz fc_chunkReplayStartTime;

	SpinLockAcquire(&fc_walrcv->mutex);
	fc_receivePtr = fc_walrcv->flushedUpto;
	SpinLockRelease(&fc_walrcv->mutex);

	fc_replayPtr = GetXLogReplayRecPtr(NULL);

	if (fc_receivePtr == fc_replayPtr)
		return 0;

	fc_chunkReplayStartTime = GetCurrentChunkReplayStartTime();

	if (fc_chunkReplayStartTime == 0)
		return -1;

	return TimestampDifferenceMilliseconds(fc_chunkReplayStartTime,
										   GetCurrentTimestamp());
}

/*
 * 返回网络延迟，单位为毫秒，请注意，这包括服务器之间任何
 * 时钟设置的差异，以及时区。
 */
int GetReplicationTransferLatency(void)
{
	WalRcvData *fc_walrcv = WalRcv;
	TimestampTz fc_lastMsgSendTime;
	TimestampTz fc_lastMsgReceiptTime;

	SpinLockAcquire(&fc_walrcv->mutex);
	fc_lastMsgSendTime = fc_walrcv->lastMsgSendTime;
	fc_lastMsgReceiptTime = fc_walrcv->lastMsgReceiptTime;
	SpinLockRelease(&fc_walrcv->mutex);

	return TimestampDifferenceMilliseconds(fc_lastMsgSendTime,
										   fc_lastMsgReceiptTime);
}
